package com.bawei.flink.tableapi;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class TableForKafka {

    public static void main(String[] args) throws Exception {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();

        TableEnvironment tEnv = TableEnvironment.create(settings);


        tEnv.executeSql("CREATE TABLE user_behavior (\n" +
                "  user_id BIGINT,\n" +
                "  item_id BIGINT,\n" +
                "  category_id BIGINT,\n" +
                "  behavior STRING,\n" +
                "  ts TIMESTAMP(3),\n" +
                " WATERMARK FOR ts AS ts - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'user_behavior',\n" +
                " 'properties.bootstrap.servers' = 'cdh1:9092,cdh2:9092,cdh3:9092',\n" +
                " 'properties.group.id' = 'testGroup',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'csv',\n" +
                " 'csv.ignore-parse-errors' = 'true',\n" +
                " 'csv.allow-comments' = 'true'\n" +
                ")");

        tEnv.sqlQuery("select * from user_behavior").execute().print();
        tEnv.execute("");
    }

}
